package mapreduce.phase4;

import java.io.IOException;
import java.util.StringTokenizer;

import mapreduce.Point;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

/**
 * To find the number of connected components and their sizes.
 * Here the vertices are grouped by root vertices 
 * @author Prac
 */
@SuppressWarnings("deprecation")
public class Map4 extends MapReduceBase implements
	Mapper<LongWritable, Text, Text, IntWritable> {
		private final static IntWritable part = new IntWritable(1);

		public void map(LongWritable key, Text value,
				OutputCollector<Text, IntWritable> output, Reporter reporter)
		throws IOException {
			String line = value.toString();
			StringTokenizer tokenizer = new StringTokenizer(line);
			assert (tokenizer.hasMoreTokens());
			int grp = Integer.parseInt(tokenizer.nextToken());
			if(grp > 0)
			{
				assert (tokenizer.hasMoreTokens());
				@SuppressWarnings("unused")
				Point p1 = Point.fromString(tokenizer.nextToken());
				assert (tokenizer.hasMoreTokens());
				Point p2 = Point.fromString(tokenizer.nextToken());
				
				Text t = new Text(p2.toString());
				part.set(1);
				output.collect(t, part);
			}
		}
	}
